/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package edu.isi.pegasus.planner.subworkflow.partitioner.implementation;

import edu.isi.pegasus.planner.subworkflow.partitioner.BackwardPartitioner;
import edu.isi.pegasus.planner.subworkflow.partitioner.DAXAbstraction;
import edu.isi.pegasus.planner.catalog.site.classes.SiteCatalogEntry;
import edu.isi.pegasus.planner.classes.ADag;
import edu.isi.pegasus.planner.classes.PegasusBag;
import edu.isi.pegasus.planner.dax.ADAG;
import edu.isi.pegasus.planner.dax.ParADAG;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Vector;

/**
 *
 * @author wchen
 */
public class BackwardPartitionerRun9 extends BackwardPartitioner{

    protected LinkedList subQueue;
    protected String fanOutJob;

    public BackwardPartitionerRun9(ADag mDag, PegasusBag mBag)
    {
        super(mDag, mBag);
        subQueue = new LinkedList();
        fanOutJob = "";
        
    }

    protected int getBigList(ArrayList jobList, ArrayList nextList, String jobID,boolean isExcluded)
    {
        int reVal = 0;
        boolean skip = false;
        int  depth = (Integer)depthMap.get(jobID);
        Vector pList = mDag.getParents(jobID);
        Vector cList = mDag.getChildren(jobID);
        if(cList.size() >  10 && !isExcluded)
        {
            int check = checkChildrenSame(jobID);
            if(check==1)
            {
                /**
                for(Iterator it = pList.iterator();it.hasNext();)
                {
                    String jobname = (String )it.next();
                    if(!nextList.contains(jobname))
                         nextList.add(jobname);
                }
                if(!jobList.contains(jobID))
                {
                    //jobList.add(jobID);
                }
                 * **/
                if(!nextList.contains(jobID))
                    nextList.add(jobID);
                if(jobList.contains(jobID))
                    jobList.remove(jobID);
                reVal = 1;
                skip = true;
                return reVal;
            }else if(check==0)
            {
                for(Iterator it = pList.iterator();it.hasNext();)
                {
                    String jobname = (String )it.next();
                    if(!nextList.contains(jobname))
                         nextList.add(jobname);
                }
                if(!jobList.contains(jobID))
                {
                    jobList.add(jobID);
                }
                skip =true;
                //reVal = 1;
                //return reVal;
            }
        }
        
        if(!skip)
        for(Iterator it = pList.iterator();it.hasNext();)
        {
            String jobname = (String )it.next();
            //edu.isi.pegasus.planner.classes.Job parent = getDagJob(jobname);
            String logicalID = jobID2jobLogicalID(jobname);
            String color = (String)mJobColor.get(logicalID);

            if(!depthMap.containsKey(jobname))
                depthMap.put(jobname, depth+1);
            if((color==null)|| (!color.equals("BLACK")))
            {
                  if(!jobList.contains(jobname))
                    {
                        subQueue.add(jobname);
                        jobList.add(jobname);
                        if(getBigList(jobList, nextList, jobname, false)==1)
                            reVal = 1;

                    }
            }



        }

    

        boolean flag = false;

        for(Iterator it = cList.iterator();it.hasNext();)
        {
            String jobname = (String )it.next();
            //edu.isi.pegasus.planner.classes.Job child = getDagJob(jobname);
            String logicalID = jobID2jobLogicalID(jobname);
            String color = (String)mJobColor.get(logicalID);
            if(!depthMap.containsKey(jobname))
                depthMap.put(jobname, depth-1);
            int i = (Integer)depthMap.get(jobname);
            if((color==null)||( !color.equals("BLACK")))
            {
                if(!jobList.contains(jobname))
                {
                    if(true )//subQueue.contains(jobname))
                    {
                        jobList.add(jobname);
                        if(getBigList(jobList, nextList, jobname,false)==1)
                            reVal= 1;
                    }
                    else
                    {
                        flag =true;
                    }


                }
            }
            else if(color.equals("BLACK") && false)//this should be if that one is different to the current site index
            {
                if(!jobList.contains(jobname))
                {
                    if( subQueue.contains(jobname))
                    {
                        jobList.add(jobname);
                        if(getBigList(jobList, nextList, jobname,false)==1)
                            reVal= 1;
                    }
                    else
                    {
                        flag =true;
                    }


                }

            }

        }


        return reVal;
    }
    protected int getMediumList(ArrayList jobList, ArrayList nextList, String jobID)
    {
        int reVal = 0;
        int  depth = (Integer)depthMap.get(jobID);

        Vector cList = mDag.getChildren(jobID);
        if (cList.size() > 10)
        {
            fanOutJob = jobID;
            jobList.remove(jobID);
            if(!nextList.contains(jobID))
                nextList.add(jobID);

            return reVal;
        }


        Vector pList = mDag.getParents(jobID);
        for(Iterator it = pList.iterator();it.hasNext();)
        {
            String jobname = (String )it.next();
            //edu.isi.pegasus.planner.classes.Job parent = getDagJob(jobname);
            String logicalID = jobID2jobLogicalID(jobname);
            String color = (String)mJobColor.get(logicalID);

            if(!depthMap.containsKey(jobname))
                depthMap.put(jobname, depth+1);

            if((color==null)|| (!color.equals("BLACK")))

            {
                  if(!jobList.contains(jobname))
                    {
                        subQueue.add(jobname);
                        jobList.add(jobname);
                        if(getMediumList(jobList, nextList, jobname)==1)
                            reVal = 1;
                    }
            }
        }
        return reVal;
    }
    protected int getSmallList(ArrayList jobList, ArrayList nextList, String jobID)
    {
        int reVal = 0;
        int  depth = (Integer)depthMap.get(jobID);
        Vector pList = mDag.getParents(jobID);
        if(pList.size() > 10)
        {
            for(Iterator it = pList.iterator();it.hasNext();)
            {
                String jobname = (String )it.next();
                edu.isi.pegasus.planner.classes.Job parent = getDagJob(jobname);
                String color ="";

                if(!mJobColor.containsKey(parent.logicalId))
                {
                    mJobColor.put(parent.logicalId, "WHITE");
                    color = "WHITE";
                }
                else
                    color = (String)mJobColor.get(parent.logicalId);

                if(!color.equals("BLACK") && !color.equals("BLACK"))
                {
                    nextList.add(parent.jobID);
                    //I think you should set it here
                    //mJobColor.put(parent.logicalId, "GRAY");
                    //depthMap.put(parent.jobID, i+1);

                }
            }

        }
        else
        {

            for(Iterator it = pList.iterator();it.hasNext();)
            {
                String jobname = (String )it.next();
                //edu.isi.pegasus.planner.classes.Job parent = getDagJob(jobname);
                String logicalID = jobID2jobLogicalID(jobname);
                String color = (String)mJobColor.get(logicalID);

                if(!depthMap.containsKey(jobname))
                    depthMap.put(jobname, depth+1);
                if((color==null)|| (!color.equals("BLACK")))

                {
                      if(!jobList.contains(jobname))
                        {
                            subQueue.add(jobname);
                            jobList.add(jobname);
                            if(getSmallList(jobList, nextList, jobname)==1)
                                reVal = 1;

                        }
                }



            }
        }
        
        return reVal;
    }
 
    private int checkChildrenSame(String jobID)
    {
        Vector vec = mDag.getChildren(jobID);
        if(vec.size()<=0)return 0;
        String childJob = (String)vec.get(0);
        ADAG childDax = (ADAG)mJob2DAX.get(jobID2jobLogicalID(childJob));
        for(Iterator it = mDag.getChildren(jobID).iterator();it.hasNext();)
        {
            String child = (String)it.next();
            ADAG newDax = (ADAG)mJob2DAX.get(jobID2jobLogicalID(child));
            if(newDax == null)continue;
            if(newDax!=childDax)
            {
                return 1;
            }

        }

        return 0;

    }

    /**
     *
     * @param jobID
     * @return 1: has multiple fanoutjob
     * @return 2: has single fanoutjob
     */
    private int checkMultipleFanOutJob(String jobID)
    {
        Vector vec = mDag.getChildren(jobID);
        if(vec.size()<=0)return 0;
        String fanOutJob = (String)vec.get(0);
        Vector pVec = mDag.getParents(fanOutJob);
        //do sampling rather check all of them
        int size = pVec.size();
        String fanInJob = "";
        for(int i = 0; i < 4; i ++)
        {
            String pJob = (String)pVec.get(i * size /4);
            while(true)
            {
                Vector ppVec = mDag.getParents(pJob);
                if(ppVec.size() <=0)
                {
                    System.out.println("Error here");
                    return 0;
                }
                pJob = (String)ppVec.get(0);
                if(mDag.getChildren(pJob).size()>10)
                {
                    if(fanInJob.equals("") ||fanInJob.equals(pJob))
                    {
                        fanInJob = pJob;
                        break;
                    }
                    else
                    {
                        return 1;
                    }
                }


            }
        }
        return 2;
    }
          
    @Override
    public void run()
    {
        getProperties();
        ParADAG dax = new ParADAG("subworkflow" + mDaxMap.size());
        DAXAbstraction bigDax = new DAXAbstraction(dax);
        Map fList= new HashMap<String, Double>();
        mDaxMap.put(bigDax, fList);
        int siteIndex = 0;
        LinkedList siteList = new LinkedList<DAXAbstraction>();
        bigDax.site = siteIndex;


        //long startTime = System.nanoTime();
        translate();

        SiteCatalogEntry se = (SiteCatalogEntry)mSiteIndex.get(siteIndex);
        double limit = Double.parseDouble(se.getEnvironmentVariable("SITESIZE")) *1e8;
        bigDax.limit = limit;
        int num2dax = 3;
        int size = mDag.getNoOfJobs();
        for(int i = 0; i < num2dax; i ++)
        {
            dax = new ParADAG("subworkflow" + mDaxMap.size());
            bigDax = new DAXAbstraction(dax);
            fList= new HashMap<String, Double>();
            mDaxMap.put(bigDax, fList);

            for(int j = i*size / num2dax; j < (i+1)*size /num2dax; j++)
            {
                edu.isi.pegasus.planner.classes.Job dagJob =
                    (edu.isi.pegasus.planner.classes.Job)mDag.vJobSubInfos.get(j);
                                    updateSizeMap(bigDax, dagJob);
                    addDAGJob(dagJob, dax);
                    bigDax.subDag.add(dagJob);

                    mJob2DAX.put(dagJob.getLogicalID(),dax);

            }
        }
            super.run();

        }

}
